package Concurrency2.Chapter08_Pattern.master_worker;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Instruction：
 * Author：@author MaLi
 */
public class Master<T extends Task, R> {
    // 存放worker
    private HashMap<String, Worker<T, R>> workers = new HashMap<>();
    // 存放任务的集合
    private LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
    // 结果集合
    protected Map<String, R> resultMap = new ConcurrentHashMap<>();

    //Master的任务调度线程
    private Thread thread = null;

    private AtomicLong sum = new AtomicLong(0);

    public Master(int workerCount) {
        for (int i = 0; i < workerCount; i++) {
            Worker<T, R> worker = new Worker<>();
            workers.put("worker" + i, worker);
        }
        thread = new Thread(() -> this.execute());
        thread.start();
    }

    //提交任务
    public void submit(T task) {
        taskQueue.add(task);
    }

    //获取worker结果处理的回调函数
    private void resultCallBack(Object o) {
        Task<R> task = (Task<R>) o;
        String taskName = "Worker:" + task.getWorkerId() + "-" + "Task:" + task.getId();
        R result = task.getResult();
        resultMap.put(taskName, result);
        sum.getAndAdd((Integer) result); //和的累加
    }

    //启动所有子任务
    public void execute() {
        for (; ; ) {
            // 分配任务
            for (Map.Entry<String, Worker<T, R>> entry : workers.entrySet()) {
                try {
                    T task = this.taskQueue.take();
                    Worker<T, R> worker = entry.getValue();
//                    worker.submit(task,resultCallBack(task));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    //获取最终结果
}
